package com.baosight.xdata.redis;

import io.lettuce.core.*;
import io.lettuce.core.cluster.ClusterClientOptions;
import io.lettuce.core.cluster.ClusterTopologyRefreshOptions;
import io.lettuce.core.cluster.RedisClusterClient;
import io.lettuce.core.cluster.api.StatefulRedisClusterConnection;
import io.lettuce.core.cluster.api.sync.RedisAdvancedClusterCommands;
import io.lettuce.core.models.stream.PendingMessage;
import redis.clients.jedis.util.Slowlog;

import java.text.MessageFormat;
import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.Set;

public class LettuceUtil {
    private static class Helper {
        private static final String KEY_SCHEMA = "yourProject:{0}";

        static RedisClusterClient redisClient;

        static StatefulRedisClusterConnection<String, String> connection;

        static final MessageFormat KEY_FORMAT = new MessageFormat(KEY_SCHEMA);

        static {
            Runtime.getRuntime().addShutdownHook(new Thread() {
                public void run() {
                    if (connection != null) {
                        connection.close();
                    }
                    if (redisClient != null) {
                        redisClient.shutdown();
                    }
                    System.out.println("redisClient shutdown in jvmHook.");
                }
            });
        }

        public static void init(Set<RedisURI> uris) {
//            long clusterTopologyRefreshMs = Long.valueOf(props.getProperty("clusterTopologyRefresh", "3000"));
            long clusterTopologyRefreshMs = 3000L;
            redisClient = RedisClusterClient.create(uris);

            ClusterTopologyRefreshOptions topologyRefreshOptions = ClusterTopologyRefreshOptions.builder()
                    //开启自适应刷新
                    .enableAdaptiveRefreshTrigger(ClusterTopologyRefreshOptions.RefreshTrigger.MOVED_REDIRECT, ClusterTopologyRefreshOptions.RefreshTrigger.PERSISTENT_RECONNECTS)
                    .enableAllAdaptiveRefreshTriggers()
                    .adaptiveRefreshTriggersTimeout(Duration.ofMillis(clusterTopologyRefreshMs))
                    //开启定时刷新,时间间隔根据实际情况修改
                    .enablePeriodicRefresh(Duration.ofMillis(clusterTopologyRefreshMs))
                    .build();
            ClusterClientOptions clientOptions = ClusterClientOptions.builder().topologyRefreshOptions(topologyRefreshOptions).autoReconnect(true).build();
            redisClient.setOptions(clientOptions);
            connection = redisClient.connect();

            System.out.println("==> Connected to Redis");
        }
    }

    public static void init(Set<RedisURI> uris) {
        Helper.init(uris);
    }

    public static String xadd(String key, Map object) {
        RedisAdvancedClusterCommands<String, String> commands = Helper.connection.sync();
        return commands.xadd(key, object);
    }

    public static List<StreamMessage<String, String>> xread(XReadArgs args, XReadArgs.StreamOffset streamOffset) {
        RedisAdvancedClusterCommands<String, String> commands = Helper.connection.sync();
        return commands.xread(args, streamOffset);
    }

    public static List<StreamMessage<String, String>> xreadGroup(String key, String consumer, String group) {
        RedisAdvancedClusterCommands<String, String> commands = Helper.connection.sync();
        return commands.xreadgroup(Consumer.from(group, consumer),
                XReadArgs.StreamOffset.lastConsumed(key));
    }

    public static List<StreamMessage<String, String>> xreadGroupWithArgs(String key, String group, String consumer, XReadArgs args) {
        RedisAdvancedClusterCommands<String, String> commands = Helper.connection.sync();
        return commands.xreadgroup(Consumer.from(group, consumer), args,
                XReadArgs.StreamOffset.lastConsumed(key));
    }


    public static void xack(String key, List<StreamMessage<String, String>> list, String group) {
        RedisAdvancedClusterCommands<String, String> commands = Helper.connection.sync();

        for (StreamMessage<String, String> message : list) {
            System.out.println(message.getId() + ":" + message.getBody());
            commands.xack(key, group, message.getId());
        }
    }

    public static void xackWithBatch(String key, String group, List<String> messageIdsList) {
        RedisAdvancedClusterCommands<String, String> commands = Helper.connection.sync();
        String[] messagesId = messageIdsList.toArray(new String[messageIdsList.size()]);
        commands.xack(key, group, messagesId);
        System.out.println(String.format("stream[%s] , group[%s], ack message size [%d]", key, group, messageIdsList.size()));
    }

    public static void xdelWithBatch(String key, List<String> messageIdsList) {
        RedisAdvancedClusterCommands<String, String> commands = Helper.connection.sync();
        String[] messagesId = messageIdsList.toArray(new String[messageIdsList.size()]);
        commands.xdel(key, messagesId);
        System.out.println(String.format("stream[%s] , del message size [%d]", key, messageIdsList.size()));
    }

    public static void xdel(String key, List<StreamMessage<String, String>> list) {
        RedisAdvancedClusterCommands<String, String> commands = Helper.connection.sync();

        for (StreamMessage<String, String> message : list) {
            System.out.println(message.getId() + ":" + message.getBody());
            commands.xdel(key, message.getId());
        }
    }

    public static String getInfo() {
        RedisAdvancedClusterCommands<String, String> commands = Helper.connection.sync();
        return commands.info();
    }

    public static String getInfo(String val) {
        RedisAdvancedClusterCommands<String, String> commands = Helper.connection.sync();
        return commands.info(val);
    }

    public static String getClusterInfo() {
        RedisAdvancedClusterCommands<String, String> commands = Helper.connection.sync();
        return commands.clusterInfo();
    }

    public static String getClusterNodes() {
        RedisAdvancedClusterCommands<String, String> commands = Helper.connection.sync();
        return commands.clusterNodes();
    }

    public static List<Object> getSlowlogGet() {
        RedisAdvancedClusterCommands<String, String> commands = Helper.connection.sync();
        return commands.slowlogGet();
    }

    public static List<Object> xinfoStream(String streamKey) {
        RedisAdvancedClusterCommands<String, String> commands = Helper.connection.sync();
        return commands.xinfoStream(streamKey);
    }

    public static List<Object> xinfoGroups(String streamKey) {
        RedisAdvancedClusterCommands<String, String> commands = Helper.connection.sync();
        return commands.xinfoGroups(streamKey);
    }

    public static List<String> keys(String streamKey) {
        RedisAdvancedClusterCommands<String, String> commands = Helper.connection.sync();
        return commands.keys(streamKey);
    }

    public static List<PendingMessage> xpendingGroup(String key, String group, String consumer, Range range, Limit limit) {
        RedisAdvancedClusterCommands<String, String> commands = Helper.connection.sync();
        return commands.xpending(key, Consumer.from(group, consumer), range, limit);
    }

    public static List<Object> getConsumerInfo(String streamKey, String group) {
        RedisAdvancedClusterCommands<String, String> commands = Helper.connection.sync();
        return commands.xinfoConsumers(streamKey, group);
    }
}
